package com.burges.net.dataStream.waterMark

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/1/27 
  * 描述     使用固定时间间隔指定Timestamps和Watermarks
  */
object StableTimeSpanTImestampAndWatermark {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment

		//指定系统时间概念为EventTime
		environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
		val input = environment.fromCollection(List(("a", 1L, 1), ("b", 1L, 1), ("b", 3L, 1)))
		input.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)](Time.seconds(10)) {
			// 定义抽取EventTime Timestamps逻辑
			override def extractTimestamp(t: (String, Long, Int)): Long = t._2
		})
		environment.execute("streaming stable time span timestamps and watermark")

	}
}
